package com.fanxuankai.boot.monitor.jdbc;

import com.fanxuankai.boot.monitor.Message;
import com.fanxuankai.boot.monitor.Monitor;
import com.fanxuankai.boot.monitor.Notifier;
import com.zaxxer.hikari.HikariDataSource;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.util.CollectionUtils;

import javax.annotation.Resource;
import javax.sql.DataSource;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.stream.Collectors;

/**
 * @author fanxuankai
 */
public class ChecksumMonitor implements Monitor {
    @Resource
    private Notifier<Message> notifier;
    @Resource
    private ChecksumProperties properties;
    @Resource
    private ScheduledThreadPoolExecutor executor;
    private final Map<String, DataSource> dataSourceCache = new HashMap<>();
    private static final String DIALECT_MYSQL = "jdbc:mysql:";
    private static final String DIALECT_ORACLE = "jdbc:oracle:thin:";

    @Override
    public String title() {
        return properties.getTitle();
    }

    @Override
    public String cron() {
        return properties.getCron();
    }

    @Override
    public void exec() throws ExecutionException, InterruptedException {
        if (CollectionUtils.isEmpty(properties.getConfigs())) {
            return;
        }
        for (ChecksumProperties.MasterSlave masterSlave : properties.getConfigs()) {
            checksum(masterSlave);
        }
    }

    private void checksum(ChecksumProperties.MasterSlave masterSlave) throws ExecutionException, InterruptedException {
        for (String schema : masterSlave.getSchemas()) {
            Future<List<Counter>> masterCountersFuture = executor.submit(() -> count(masterSlave, masterSlave.getMaster(), schema));
            List<Future<List<Counter>>> slaveCountersFutures = masterSlave.getSlaves()
                    .stream()
                    .map(slave -> executor.submit(() -> count(masterSlave, slave, schema)))
                    .collect(Collectors.toList());
            List<Counter> masterCounters = masterCountersFuture.get();
            List<Map<String, Integer>> slaveCountMaps = new ArrayList<>(masterSlave.getSlaves().size());
            for (Future<List<Counter>> slaveCounterFuture : slaveCountersFutures) {
                slaveCountMaps.add(slaveCounterFuture.get()
                        .stream()
                        .collect(Collectors.toMap(Counter::getTableName, Counter::getCount)));
            }
            Map<String, List<Integer>> diff = masterCounters.stream()
                    .filter(masterCounter -> {
                        int threshold = Optional.ofNullable(masterSlave.getThreshold().get(schema))
                                .map(o -> o.getOrDefault(masterCounter.getTableName(), properties.getGlobalThreshold()))
                                .orElse(properties.getGlobalThreshold());
                        return slaveCountMaps.stream().anyMatch(slaveCountMap -> {
                            Integer slaveCount = Optional.ofNullable(slaveCountMap.get(masterCounter.getTableName()))
                                    .orElse(0);
                            return Math.abs(masterCounter.getCount() - slaveCount) > threshold;
                        });
                    }).collect(Collectors.toMap(Counter::getTableName, masterCounter -> {
                        List<Integer> countList = new ArrayList<>();
                        countList.add(masterCounter.getCount());
                        countList.addAll(slaveCountMaps.stream()
                                .map(o -> Optional.ofNullable(o.get(masterCounter.getTableName()))
                                        .orElse(0))
                                .collect(Collectors.toList()));
                        return countList;
                    }));
            if (!diff.isEmpty()) {
                Message message = new Message();
                message.setTitle(properties.getTitle());
                message.setSubtitle(Optional.ofNullable(properties.getSubtitle())
                        .orElseGet(() -> {
                            List<String> jdbcUrls = new ArrayList<>();
                            jdbcUrls.add(masterSlave.getMaster().getJdbcUrl());
                            masterSlave.getSlaves()
                                    .stream()
                                    .map(ChecksumProperties.DataBase::getJdbcUrl)
                                    .forEach(jdbcUrls::add);
                            return String.join(" -> ", jdbcUrls);
                        }));
                message.setContents(diff.entrySet().stream()
                        .map(entry -> String.format("%s.%s: %s", schema, entry.getKey(), entry.getValue()
                                .stream()
                                .map(String::valueOf)
                                .collect(Collectors.joining(" -> "))))
                        .collect(Collectors.toList()));
                message.setDescription("共计 " + diff.size() + " 张表的数据不一致");
                notifier.notice(message);
            }
        }
    }

    private List<Counter> count(ChecksumProperties.MasterSlave masterSlave,
                                ChecksumProperties.DataBase dataBase,
                                String schema) {
        DataSource dataSource = dataSourceCache.computeIfAbsent(dataBase.getJdbcUrl(), s -> {
            HikariDataSource ds = new HikariDataSource();
            ds.setJdbcUrl(dataBase.getJdbcUrl());
            ds.setUsername(dataBase.getUsername());
            ds.setPassword(dataBase.getPassword());
            return ds;
        });
        JdbcTemplate template = new JdbcTemplate(dataSource);
        Set<String> tableNames = Collections.emptySet();
        if (CollectionUtils.isEmpty(masterSlave.getIncludeTables().get(schema))) {
            if (dataBase.getJdbcUrl().contains(DIALECT_MYSQL)) {
                String sql = String.format("SELECT TABLE_NAME table_name FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_SCHEMA = '%s'", schema);
                tableNames = new HashSet<>(template.queryForList(sql, String.class));
            } else if (dataBase.getJdbcUrl().contains(DIALECT_ORACLE)) {
                String sql = String.format("SELECT TABLE_NAME table_name FROM all_tables WHERE OWNER = '%s'", schema);
                tableNames = new HashSet<>(template.queryForList(sql, String.class));
            }
        } else {
            tableNames = masterSlave.getIncludeTables().get(schema);
        }
        if (!CollectionUtils.isEmpty(masterSlave.getExcludeTables().get(schema))) {
            tableNames.removeAll(masterSlave.getExcludeTables().get(schema));
        }
        if (tableNames.isEmpty()) {
            return Collections.emptyList();
        }
        return template.query(tableNames.stream()
                .map(s -> String.format("select '%s' table_name, count(*) count from %s.%s", s, schema, s))
                .collect(Collectors.joining(" union all ")), BeanPropertyRowMapper.newInstance(Counter.class));
    }

    private static class Counter {
        private String tableName;
        private Integer count;

        public String getTableName() {
            return tableName;
        }

        public void setTableName(String tableName) {
            this.tableName = tableName;
        }

        public Integer getCount() {
            return count;
        }

        public void setCount(Integer count) {
            this.count = count;
        }
    }
}
